package com.atguigu.edu.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.TableProcess;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcess, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Map<String,TableProcess> configMap = new HashMap<>();
    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor){
        this.mapStateDescriptor = mapStateDescriptor;
    }

    public void open(Configuration parameters) throws Exception {
        //从配置表将配置信息读取出来  放到集合中进行预加载
        //注册驱动
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop102:3306/edu_config?" +
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        //获取数据库操作对象
        String sql = "select * from edu_config.table_process_dim ";
        PreparedStatement ps = conn.prepareStatement(sql);
        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        //处理结果集
        ResultSetMetaData metaData = rs.getMetaData();
        while (rs.next()){
            //定义一个新的json对象，用于封装查询出来的一条记录
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName,columnValue);
            }
            //将json对象转换为实体类对象
            TableProcess tableProcess = jsonObj.toJavaObject(TableProcess.class);
            //将读取到的这条配置信息放到configMap中
            configMap.put(tableProcess.getSourceTable(),tableProcess);
        }
        //释放资源
        rs.close();
        ps.close();
        conn.close();
    }

    @Override
    public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        //获取处理业务数据表表名
        String tableName = jsonObj.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        //根据业务数据库表表名到广播状态中获取对应的配置信息
        TableProcess tableProcess = null;

        //判断配置信息是否存在
        if((tableProcess = broadcastState.get(tableName)) != null
                ||(tableProcess = configMap.get(tableName)) != null){
            //如果从广播状态中获取到的配置信息不为空，说明当前处理的是维度数据
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            //在向下游传递数据前，过滤掉不需要传递的属性
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj,sinkColumns);
            //在向下游传递数据前，补充操作类型
            String type = jsonObj.getString("type");
            dataJsonObj.put("type",type);
            //在向下游传递数据前，补充当前维度表对应的配置对象
            dataJsonObj.put("table_process",tableProcess);
            //将维度数据传递到下游
            collector.collect(dataJsonObj);
        }
    }

    //过滤掉不需要的数据
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        String[] columnArr = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columnArr);
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();
        entrySet.removeIf(entry-> !columnList.contains(entry.getKey()));
    }

    @Override
    public void processBroadcastElement(TableProcess tableProcess, BroadcastProcessFunction<JSONObject, TableProcess, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
        //获取对配置表的操作类型
        String op = tableProcess.getOp();
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        //获取配置表中配置业务数据库维度表表名
        String sourceTable = tableProcess.getSourceTable();
        //判断是否是删除操作
        if("d".equals(op)){
            //将对应的配置信息从广播状态中删除掉
            broadcastState.remove(sourceTable);
            configMap.remove(sourceTable);
        }else{
            //如果对配置表做了删除外的其它操作，将对应的配置信息放到或者更新到广播状态中
            broadcastState.put(sourceTable,tableProcess);
            configMap.put(sourceTable,tableProcess);
        }
    }
}
